package cn.cmft.jc.transform;

import cn.cmft.jc.transform.parse.*;
import cn.cmft.jc.transform.parse.strategy.ChangeOptionDDLParseContext;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.RenameTableStatement;
import net.sf.jsqlparser.statement.drop.Drop;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;


public class ExtractFromSchemaChangeData extends RichMapFunction<String, Optional<ChangeOptionMessage>> {
    private final Logger logger = LoggerFactory.getLogger(ExtractFromSchemaChangeData.class);
    private final String host;
    private final int port;
    private static final String HISTORY_RECORD = "historyRecord";


    public ExtractFromSchemaChangeData(ParameterTool tool) {
        this.host = tool.get("monitor.datasource.hostname");
        this.port = tool.getInt("monitor.datasource.port");
    }
    @Override
    public Optional<ChangeOptionMessage> map(String record) {
        JSONObject jsonObject = JSONObject.parseObject(record);
        // 预先过滤掉一些不需要处理的数据
        if (!filterInvalidData(jsonObject)){
            return Optional.empty();
        }

        // 获取公共属性
        JSONObject source = jsonObject.getJSONObject("source");
        JSONObject historyRecord = JSON.parseObject(jsonObject.getString(HISTORY_RECORD));
        String db =  source.getString("db");
        String table = source.getString("table");
        Long tsMs = source.getLong("ts_ms");
        String ddl = historyRecord.getString("ddl").replaceAll("\\r\\n"," ");

        // 获取变更的详情
        ChangeOptionDDLParseContext ddlParseContext = new ChangeOptionDDLParseContext();
        Map<ChangeOperationType, List<ChangeDetail>> changeOperationDetails = ddlParseContext.executeParse(ddl);

        // 解析完ddl所有语句，返回相应信息
        ChangeOptionMessage changeOptionMessage = new ChangeOptionMessage.Builder()
                .host(host)
                .port(port)
                .database(db)
                .table(table)
                .tmMs(tsMs)
                .ddl(ddl)
                .changeOperations(changeOperationDetails)
                .build();
        return Optional.of(changeOptionMessage);
    }
    private boolean filterInvalidData(JSONObject jsonObject){
        if (!jsonObject.containsKey(HISTORY_RECORD)){
            logger.info("监控记录不是ddl操作，不做处理...");
            return false;
        }
        JSONObject historyRecord = JSON.parseObject(jsonObject.getString(HISTORY_RECORD));
        JSONArray tableChanges = historyRecord.getJSONArray("tableChanges");

        try {
            Statement ddlStatement = CCJSqlParserUtil.parse(historyRecord.getString("ddl"));
            return (ddlStatement instanceof Alter &&  !tableChanges.isEmpty() )||
                    ddlStatement instanceof Drop ||
                    ddlStatement instanceof RenameTableStatement;
        } catch (JSQLParserException e) {
            logger.warn("DDL语句解析错误: \n", e);
        }
        return false;
    }

}
